package org.fight.landowner.remoting.netty;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;
import org.fight.landowner.common.DelayExecuteTaskQueue;
import org.fight.landowner.common.DelayTask;
import org.fight.landowner.common.serialize.BytesSerializableFactory;
import org.fight.landowner.common.serialize.SerializableType;
import org.fight.landowner.remoting.*;
import org.fight.landowner.remoting.exception.RemotingRequestExceptionFactory;
import org.fight.landowner.remoting.exception.RemotingRequestExceptionType;
import org.fight.landowner.remoting.exception.RequestTimeoutException;
import org.fight.landowner.remoting.exception.SendMessageException;
import org.fight.landowner.remoting.protocol.RemotingCommand;
import org.fight.landowner.remoting.protocol.RemotingCommandHeader;
import org.fight.landowner.remoting.protocol.RemotingCommandType;
import sun.reflect.generics.reflectiveObjects.ParameterizedTypeImpl;

import java.lang.reflect.Type;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;

/**
 * @Author wangxuehao02
 * @Date 2024/2/1 8:35 PM
 * @Description
 **/

@Slf4j
public abstract class NettyAbstractRemotingService extends AbstractRemotingService {

    // 处理器与执行器对应的执行线程池
    protected Map<Short, Pair<CommandHandler, ExecutorService>> handler2executorServiceMap = new HashMap<>();

    // 响应表
    protected ConcurrentHashMap<Integer/*requestId*/, ResponseFuture> responseTable = new ConcurrentHashMap<>();

    // 响应延迟队列用于清除过期 ResponseFuture
    protected DelayExecuteTaskQueue responseDelayQueue;

    /**
     * 注册处理器
     * @param code
     * @param handler
     * @param executorService
     */
    public void registerHandler(short code, CommandHandler handler, ExecutorService executorService) {
        handler2executorServiceMap.put(code, Pair.of(handler, executorService));
    }

    /**
     * 处理请求
     * @return
     */
    public void processRequest(Channel channel, RemotingCommand remotingCommand) {
        RemotingCommandHeader requestHeader = remotingCommand.getHeader();

        short code = requestHeader.getCode();

        Pair<CommandHandler, ExecutorService> handlerExecutorServicePair = handler2executorServiceMap.get(code);

        RemotingCommand responseRemotingCommand = new RemotingCommand();
        RemotingCommandHeader remotingCommandHeader = RemotingCommandHeader.buildResponseRemotingCommandHeader();
        responseRemotingCommand.setHeader(remotingCommandHeader);

        if (handlerExecutorServicePair == null) {
            // todo
            remotingCommandHeader.setCode((short) RemotingRequestExceptionType.UNKNOWN_REQUEST_TYPE.getCode());
            channel.writeAndFlush(remotingCommandHeader);
        } else {
            ExecutorService executorService = handlerExecutorServicePair.getRight();
            CommandHandler commandHandler = handlerExecutorServicePair.getLeft();

            executorService.execute(() -> {
                try {
                    Object responseResult = commandHandler.handlerRequest(remotingCommand);

                    if (responseResult != null) {
                        remotingCommandHeader.setCode(code);
                        remotingCommandHeader.setRequestId(requestHeader.getRequestId());
                        responseRemotingCommand.setHeader(remotingCommandHeader);

                        responseRemotingCommand.setData(BytesSerializableFactory.serialize(responseResult, requestHeader.getSerializableProtocol()));

                        channel.writeAndFlush(responseRemotingCommand);
                    }
                    // todo 怎么处理同步异步是否需要等待？
                    /**
                     * 1. 判断是请求/响应，响应无须返回，请求判断其类型，需要加标记
                     *
                     * 返回的 code 码和请求的 code 码是否需要是同一个？
                     * 返回的消息需不需要定期删除，
                     * 不删除的话如果网络 io 时间大于等待时间会不会产生脏数据？如何定期清理呢？
                     *
                     * 加上两个标签
                     * 1. 请求是否需要返回
                     * 2. 同步返回，返回都需要异步
                     */
                } catch (Exception e) {
                    // todo
                    remotingCommandHeader.setCode((short) RemotingRequestExceptionType.SYSTEM_PROCESSING_EXCEPTION.getCode());
                    channel.writeAndFlush(remotingCommandHeader);
                }
            });
        }
    }

    /**
     * 同步响应
     * 异步响应
     * 异常
     * todo 如果发起请求但是请求丢失了/响应丢失了，清理 responseTable
     * @param channel
     * @param remotingCommand
     */
    public void processResponse(Channel channel, RemotingCommand remotingCommand) {
        RemotingCommandHeader header = remotingCommand.getHeader();

        short code = header.getCode();
        SerializableType serializableProtocol = header.getSerializableProtocol();

        int requestId = header.getRequestId();

        ResponseFuture responseFuture = responseTable.get(requestId);

        if (responseFuture != null) {
            responseFuture.setSuccess(code == 0);
            Class responseClazz = responseFuture.getResponseClazz();
            RemotingRequestType requestType = responseFuture.getRequestType();
            SendCallback sendCallback = responseFuture.getSendCallback();

            if (responseFuture.isSuccess()) {
                switch (requestType) {
                    case SYNC: {
                        responseFuture.putAndRouse(BytesSerializableFactory.deSerialize(remotingCommand.getData(), responseClazz, serializableProtocol));
                        break;
                    }
                    case ASYNC: {
                        responseFuture = responseTable.remove(requestId);

                        if (responseFuture != null) {
                            // todo 获取范型类型，不够优雅，后续优化
                            Type type = Object.class;

                            Type[] genericInterfaces = sendCallback.getClass().getGenericInterfaces();

                            if (genericInterfaces != null && genericInterfaces.length > 0) {
                                Type genericInterface = genericInterfaces[0];

                                Type[] actualTypeArguments = ((ParameterizedTypeImpl) genericInterface).getActualTypeArguments();
                                if (actualTypeArguments != null && actualTypeArguments.length > 0) {
                                    type = actualTypeArguments[0];
                                }
                            }

                            sendCallback.callback(BytesSerializableFactory.deSerialize(remotingCommand.getData(), (Class)type, serializableProtocol), null);
                        }
                        break;
                    }
                }
            } else {
                // todo 待制定统一返回码
                /**
                 * 网络协议的返回码 例如 http 协议 200 404 ...
                 * 500 服务器异常
                 * 400 不能处理的请求
                 */
                switch (requestType) {
                    case SYNC: {
                        responseFuture.setCause(RemotingRequestExceptionFactory.getRemotingRequestExceptionByCode(code));
                        responseFuture.rouse();
                        break;
                    }
                    case ASYNC: {
                        responseTable.remove(responseFuture);
                        sendCallback.callback(null, RemotingRequestExceptionFactory.getRemotingRequestExceptionByCode(code));
                        break;
                    }
                }
            }
        }
    }

    public void processRemotingCommand(Channel channel, RemotingCommand remotingCommand) {
        RemotingCommandHeader header = remotingCommand.getHeader();

        RemotingCommandType commandType = header.getType();

        switch (commandType) {
            case REQUEST: {
                processRequest(channel, remotingCommand);
                break;
            }
            case RESPONSE: {
                processResponse(channel, remotingCommand);
                break;
            }
        }
    }

    public void asyncSend(Channel channel, RemotingCommand remotingCommand, long timeoutMillis, SendCallback sendCallback) {
        RemotingCommandHeader header = remotingCommand.getHeader();

        int requestId = header.getRequestId();
        try {
            ResponseFuture responseFuture = new ResponseFuture();
            responseFuture.setRequestType(RemotingRequestType.ASYNC);
            responseFuture.setSendCallback(sendCallback);
            // 预留 1s 的 buffer
            responseDelayQueue.offer(new ResponseCleanDelayTask(timeoutMillis + 1000, requestId));

            responseTable.put(requestId, responseFuture);

            channel.writeAndFlush(remotingCommand).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) {
                    if (!future.isSuccess()) {
                        ResponseFuture curResponseFuture = responseTable.remove(requestId);

                        if (curResponseFuture != null) {
                            curResponseFuture.getSendCallback().callback(null, new SendMessageException(future.cause()));
                        }
                    }
                }
            });
        } catch (Exception e) {
            ResponseFuture curResponseFuture = responseTable.remove(requestId);
            log.error("asyncSend error");

            if (curResponseFuture != null) {
                throw new SendMessageException(e.getMessage(), e.getCause());
            }
        }
    }

    /**
     * todo 需不需要不加超时时间的同步方法，如果不加的话会不会永久阻塞，今儿导致内存泄漏，线程过多？？？
     * @param channel
     * @param remotingCommand
     * @param timeoutMillis
     */
    public <T> T syncSend(Channel channel, RemotingCommand remotingCommand, long timeoutMillis, Class<T> responseClazz) throws RequestTimeoutException {
        RemotingCommandHeader header = remotingCommand.getHeader();

        int requestId = header.getRequestId();

        try {
            ResponseFuture responseFuture = new ResponseFuture<>();
            responseFuture.setRequestType(RemotingRequestType.SYNC);
            responseFuture.setResponseClazz(responseClazz);

            responseTable.put(requestId, responseFuture);

            long startTimeMillis = System.currentTimeMillis();

            channel.writeAndFlush(remotingCommand).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) {
                    if (!future.isSuccess()) {
                        responseFuture.setSuccess(false);
                        responseFuture.setCause(future.cause());
                        // 发送失败立即唤醒
                        responseFuture.rouse();
                    }
                }
            });

            // todo 超时后如何抛出超时异常
            // todo 超时等待应该放在外面，writeAndFlush & 其他逻辑也可能会耗费大量时间
            Object result = responseFuture.get(timeoutMillis);

            if (!responseFuture.isSuccess()) {
                throw new SendMessageException(responseFuture.getCause());
            }

            if (System.currentTimeMillis() - startTimeMillis > timeoutMillis) {
                throw new RequestTimeoutException();
            }

            return (T) result;
        } catch (RequestTimeoutException | SendMessageException e) {
            throw e;
        } catch (Exception e) {
            log.error("syncSend error");
            throw new SendMessageException(e);
        } finally {
            responseTable.remove(requestId);
        }
    }

    /**
     * 无响应发送
     */
    public void noResponseSend(Channel channel, RemotingCommand remotingCommand) {
        try {
            channel.writeAndFlush(remotingCommand);
        } catch (Exception e) {
            log.error("unresponsiveSend error");
            throw new SendMessageException(e);
        }
    }

    /**
     * 处理过期请求
     * todo 应该使用异步避免阻塞
     */
    protected void handlerTimeoutRequest(Integer requestId) {
        ResponseFuture responseFuture = responseTable.remove(requestId);

        if (responseFuture != null) {
            SendCallback sendCallback = responseFuture.getSendCallback();
            sendCallback.callback(null, new RequestTimeoutException());
        }
    }
}
